package com.atbeijing.bigdata.spark.core.rdd.operator.action

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark10_Oper_Action {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("ActionOperator")
        val sc = new SparkContext(conf)

        // TODO 算子 - 行动
        val rdd = sc.makeRDD(
            List("Hello", "Hive", "Hadoop", "Scala", "Spark")
        )

        //rdd.filter(_.startsWith("H")).collect().foreach(println)
        val search = new Search("H");
        search.findDatas1(rdd).collect().foreach(println)
//        search.findDatas(rdd).collect().foreach(println)
        sc.stop()

    }
    // 样例类默认实现可序列化接口,class前加case即可
    class Search( query:String ) {
        def findDatas(rdd : RDD[String]): RDD[String] = {
            //Exception in thread "main" org.apache.spark.SparkException: Task not serializable
            //rdd使用了当前对象(调用该方法的对象)的属性query,所以对象要序列化,以便传输
            rdd.filter(s=>s.contains(this.query))
        }
        def findDatas1(rdd : RDD[String]): RDD[String] = {
            val q: String = query
            //将query传给了q,q不是当前对象的属性,所以当前对象不需要序列化
            // q是String类型对象,String实现了序列化,所以可以直接传输
            rdd.filter(s=>s.contains(q))
        }
    }

}
